# coding: utf-8
"""
流程说明：通过filebeat收集异常日志，推送至logstash，logstash在将数据推送到kafka中，该脚本修改kafka数据进行处理
脚本主要功能说明：
1、通过数据库和配置文件可以将异常日志推送给具体负责人
2、可根据不同产品线自定义函数列，可以更个性化的处理日志、告警等行为
"""

import re
import json
import time
import requests
import queue
import threading
import kafka
import traceback
import pymysql
import logging
import datetime

from flask import Flask, request, jsonify

logging.basicConfig(filename='./ding_ding_alert.log',
                    format='%(asctime)s - %(name)s - %(levelname)s -%(module)s:  %(message)s',
                    datefmt='%Y-%m-%d %H:%M:%S',
                    level=logging.INFO)

app = Flask(__name__)
User_Alert_Dict = {}
Basic_Alert_Dict = {}
Sms_Alert_Date_Dict = {}

def generateAlertDict():
    try:
        connection = pymysql.connect(
            host='',
            port=3306,
            user='',
            password='',
            database='')
        _sql = """
            SELECT
                biz_code,
                importance,
                responsible,
                mobile 
            FROM
                log_config 
            WHERE
                deleted_id = 0;
        """
        with connection.cursor() as cursor:
            cursor.execute(_sql)
            columns = [col[0] for col in cursor.description]
            queryResults = [
                dict(zip(columns, row))
                for row in cursor.fetchall()
            ]
            for item in queryResults:
                User_Alert_Dict[item['biz_code']] = item
        logging.info("已完成用户告警字典的刷新")
    except:
        logging.error('generateAlertDict 刷新用户告警字典失败')
        logging.error(traceback.format_exc())


def json_generate_alert_dict():
    try:
        json_path = r'/home/py_project/errorLog_push/user_dict.json'
        with open(json_path, 'r') as f:
            data = json.load(f)
        for k, v in data.items():
            Basic_Alert_Dict[k] = v
        logging.info("已完成基础用户告警字典的刷新")
    except:
        logging.error('json_generate_alert_dict 刷新基础用户告警字典失败')
        logging.error(traceback.format_exc())

def date_list_generate(date_list):
    date_new_list = []
    for d in date_list:
        date_split = d.split('-')
        if len(date_split) == 1:
            date_new_list.append(d)
        elif len(date_split) == 2:
            dateStart = datetime.datetime.strptime(date_split[0], '%Y%m%d')
            dateEnd = datetime.datetime.strptime(date_split[1], '%Y%m%d')
            date_new_list.append(dateStart.strftime('%Y%m%d'))
            while dateStart < dateEnd:
                dateStart += datetime.timedelta(days=1)
                date_new_list.append(dateStart.strftime('%Y%m%d'))
        else:
            print('日期列表生成失败')
    return date_new_list

def sms_alert_date_dict():
    try:
        json_path = r'/home/py_project/errorLog_push/params_dict.json'
        with open(json_path, 'r') as f:
            data = json.load(f)
        for k, v in data.items():
            if k == "holidays_exception":
                Sms_Alert_Date_Dict['holidays_exception'] = date_list_generate(v)
            elif k == "workdays_exception":
                Sms_Alert_Date_Dict['workdays_exception'] = date_list_generate(v)
        logging.info("已完成日期字典的刷新")
    except:
        logging.error('sms_alert_date_dict 刷新日期字典失败')
        logging.error(traceback.format_exc())


@app.route('/refreshDict', methods=['GET'])
def refreshDict():
    try:
        logging.info("请求刷新用户告警字典")
        generateAlertDict()
    except:
        print('refreshDict 刷新用户告警字典失败')
        print(traceback.format_exc())
    finally:
        resp = 'success'
        return resp


@app.route('/refreshBasicDict', methods=['GET'])
def refreshBasicDict():
    try:
        logging.info("请求刷新基础用户告警字典")
        json_generate_alert_dict()
    except:
        print('refreshBasicDict 刷新基础用户告警字典失败')
        print(traceback.format_exc())
    finally:
        resp = 'success'
        return resp

@app.route('/refreshDateDict', methods=['GET'])
def refreshDateDict():
    try:
        logging.info("请求刷新日期字典")
        sms_alert_date_dict()
    except:
        print('refreshDateDict 刷新日期字典失败')
        print(traceback.format_exc())
    finally:
        resp = 'success'
        return resp

class DDingPush(threading.Thread):
    def __init__(self, d_url=None, d_queue=None, d_hook_list=None):
        threading.Thread.__init__(self)
        self.pattern_source = re.compile(r'.*\[业务来源:(.*?)\].*')
        self.pattern_module = re.compile(r'.*\[代码模块:(.*?)\].*')
        self.pattern_method = re.compile(r'.*?\].*?\] (.*?) .*')

        self.template = {
            "msgtype": "markdown",
            "markdown": {
                "title": "接收服务器异常",
                "text": "接收服务器异常，无法正常推送异常告警"
            },
            "at": {
                'atMobiles': [],
                "isAtAll": True
            }
        }
        self.sms_template = {
            "paramList": [], # ['模块', '来源', '次数']
            "mobile": "",
            "modelId": "NM124"
        }
        self.dding_url = d_url
        self.dding_url_number = len(d_url.keys())
        self.project_name_list = []
        # 频率
        self.frequency_dict = {}

        self.frequency_dict_sms = {}
        # 运行的钩子
        self.hook_func_list = ['frequency_control', 'save_error_log', 'alert_data_generation', 'ding_ding_push'] if \
            d_hook_list is None else d_hook_list

        self.data_queue = d_queue
        self.next_index = 0
        self.sms_push_url = ''
        self.displayUrl = ''
        self.submitUrl = ''
        self.alertDict = {
            'ip': '',
            'project': '',
            'msg': '',  # 最大200个字符
            'stamp': '',
            'alert_dict': {}
        }

    @staticmethod
    def url_hash(project_index, url_number):
        """
        基于项目名在列表中的位置来固定命中相同的url
        """
        return int(project_index % url_number)

    @staticmethod
    def is_holiday():
        day = datetime.datetime.today()
        week_day = datetime.datetime.weekday(day) + 1
        is_work_day_in_week = week_day in range(1, 6)
        day_str = f'{day.year}{str(day.month).zfill(2)}{str(day.day).zfill(2)}'

        if day_str in Sms_Alert_Date_Dict['workdays_exception']:
            return False
        elif day_str in Sms_Alert_Date_Dict['holidays_exception']:
            return True
        elif is_work_day_in_week:
            return False
        else:
            return False

    def frequency_control(self):
        timestamp = int(time.time())
        method_obj = self.pattern_method.findall(self.alertDict['complete_msg'])
        if len(method_obj) > 0:
            method_name = method_obj[0]
            if method_name in self.frequency_dict.keys():
                self.frequency_dict[method_name] = [i for i in self.frequency_dict[method_name] if timestamp - i < 60]
                self.frequency_dict[method_name].append(timestamp)
            else:
                self.frequency_dict[method_name] = [timestamp]

            if len(self.frequency_dict[method_name]) > 5:
                logging.info('{}项目的{}类型告警日志超出限制，不进行推送'.format(self.alertDict['project'], method_name))
                return False
            else:
                return True
        else:
            return True

    def frequency_control_sms(self, f_number=5, f_silence_time=60):
        if self.alertDict['alert_dict'] != {}:
            timestamp = int(time.time())
            if 'biz_code_list' in self.alertDict['alert_dict']:
                key_name = '_'.join(self.alertDict['alert_dict']['biz_code_list'])
                if key_name in self.frequency_dict_sms.keys():
                    if self.frequency_dict_sms[key_name]['number'] >= f_number:
                        if int(timestamp - self.frequency_dict_sms[key_name]['timestamp']) > f_silence_time:
                            self.frequency_dict_sms[key_name]['number'] = 1
                    else:
                        self.frequency_dict_sms[key_name]['number'] += 1
                        self.frequency_dict_sms[key_name]['timestamp'] = timestamp
                        if self.frequency_dict_sms[key_name]['number'] == f_number:
                            return True
                else:
                    self.frequency_dict_sms[key_name] = {
                        'number': 1,
                        'timestamp': timestamp
                    }
            return False
        else:
            return False

    def __next__(self, project_name):
        if project_name not in self.project_name_list:
            self.project_name_list.append(project_name)

        dd_key = self.url_hash(self.project_name_list.index(project_name), self.dding_url_number)
        url = self.dding_url[dd_key]
        return url

    def save_error_log(self):
        self.alertDict['stamp'] = time.time()
        params = {
            'project': self.alertDict['project'] + '-' + self.alertDict['ip'],
            'stamp': self.alertDict['stamp'],
            'context': self.alertDict['complete_msg']
        }
        try:
            # logging.info("落库数据: {}".format(params))
            r = requests.post(self.submitUrl, data=params)
            if r.status_code == requests.codes.ok:
                logging.info("数据写入成功")
                return True
            else:
                logging.error("数据写入失败: {}".format(r.json()))
                return True
        except:
            logging.error('数据落库失败')
            logging.error(traceback.format_exc())
            return True

    def generateAlert(self):
        jump_url = self.displayUrl.format(self.alertDict['project'] + '-' + self.alertDict['ip'],
                                          self.alertDict['stamp'])
        self.template['markdown']['title'] = '{} {}'.format(self.alertDict['project'], self.alertDict['ip'])

        if self.alertDict['alert_dict'] == {}:
            logging.info('生成一般告警推送信息: {}'.format(Basic_Alert_Dict.keys()))
            if self.alertDict['project'] in Basic_Alert_Dict.keys():
                self.template['at']['atMobiles'] = [Basic_Alert_Dict[self.alertDict['project']]]
                self.template['markdown']['text'] = "@{} \n\n".format(Basic_Alert_Dict[self.alertDict['project']]) + \
                                                    self.alertDict['msg'] + '\n' + \
                                                    str('[{}]({})'.format(
                                                        self.alertDict['project'] + '-' + self.alertDict['ip'],
                                                        jump_url))
                self.template['at']['isAtAll'] = False
            else:
                self.template['markdown']['text'] = self.alertDict['msg'] + '\n' + str(
                    '[{}]({})'.format(self.alertDict['project'] + '-' + self.alertDict['ip'], jump_url))
                self.template['at']['atMobiles'] = []
                self.template['at']['isAtAll'] = True
        else:
            logging.info('生成特别告警推送信息: {}'.format(self.alertDict['project']))
            mobile_list = []
            mobile_str = ''
            importance_str = ''
            for key, value in self.alertDict['alert_dict'].items():
                if key == "mobile_list":
                    mobile_list = value
                    for phone in mobile_list:
                        mobile_str += str("@{} \n\n".format(phone))
                elif key == "source" or key == "module":
                    importance_str += "[{}告警级别:{}]\n".format(key, value['importance'])

            self.template['at']['atMobiles'] = mobile_list
            self.template['markdown']['text'] = mobile_str + importance_str + '\n' + self.alertDict['msg'] + '\n' + \
                                                str('[{}]({})'.format(
                                                    self.alertDict['project'] + '-' + self.alertDict['ip'], jump_url))
            self.template['at']['isAtAll'] = False

        return json.dumps(self.template)

    def generate_sms_alert(self):
        # for key, value in self.alertDict['alert_dict'].items():
        sms_notice_dict = {}

        if self.alertDict['alert_dict'] != {}:
            mobile_list = self.alertDict['alert_dict'].get('mobile_list', None)
            source_name = self.alertDict['project'] if self.alertDict['alert_dict'].get('source', None) is None else \
                self.alertDict['alert_dict']['source']['biz_code']
            module_name = self.alertDict['project'] if self.alertDict['alert_dict'].get('module', None) is None else \
                self.alertDict['alert_dict']['module']['biz_code']

            if mobile_list is not None:
                for phone in mobile_list:
                    sms_notice_dict[phone] = [source_name, module_name, 5]
            else:
                logging.error("短信通知信息生成过程中未发现有效的号码列表")
        return sms_notice_dict

    def send_alert(self, num=0):
        header = {'Content-Type': 'application/json'}
        data = {
            "msgtype": "markdown",
            "markdown": {
                "title": "接收服务器异常",
                "text": "接收服务器异常，无法正常推送异常告警"
            },
            "at": {
                "isAtAll": True
            }
        }
        res = requests.post(self.__next__('errorAlert'),
                            headers=header,
                            data=json.dumps(data))
        if res.ok and res.json()["errcode"] == 0:
            logging.info('数据推送成功')
        else:
            logging.info('数据推送失败')


    def ding_ding_push(self, num=0):
        logging.info('开始推送钉钉通知')
        header = {'Content-Type': 'application/json'}
        data = self.generateAlert()
        res = requests.post(self.__next__(self.alertDict['project']),
                            headers=header,
                            data=data)
        if res.ok and res.json()["errcode"] == 0:
            logging.info('钉钉通知推送成功')
        else:
            logging.error('钉钉通知推送失败: {}'.format(res.json()))
            logging.error('钉钉通知推送失败数据: {}'.format(data))
        return True

    def sms_push(self):
        logging.info('开始推送SMS通知')
        header = {'Content-Type': 'application/json'}
        sms_notice_dict = self.generate_sms_alert()
        for k, v in sms_notice_dict.items():
            self.sms_template['mobile'] = k
            self.sms_template['paramList'] = v
            sms_params = json.dumps(self.sms_template)
            res = requests.post(self.sms_push_url,
                                headers=header,
                                data=sms_params)
            if res.ok:
                logging.info('SMS通知推送成功: {}'.format(res.text))
            else:
                logging.error('SMS通知推送失败: {}'.format(res.text))
                logging.error('SMS通知推送失败数据: {}'.format(sms_params))

    def alert_data_generation(self):
        self.alertDict['alert_dict'] = {}
        try:
            source_obj = self.pattern_source.findall(self.alertDict['complete_msg'])
            module_obj = self.pattern_module.findall(self.alertDict['complete_msg'])
            biz_code_list = []
            mobile_list = []
            if len(source_obj) > 0:
                biz_code = source_obj[0]
                if biz_code in User_Alert_Dict.keys():
                    biz_code_value = User_Alert_Dict[biz_code]
                    self.alertDict['alert_dict']['source'] = {
                        'importance': biz_code_value['importance'],
                        'biz_code': biz_code
                    }

                    for phone in biz_code_value['mobile'].split(','):
                        if phone not in mobile_list:
                            mobile_list.append(phone)

                    biz_code_list.append(biz_code)

            if len(module_obj) > 0:
                biz_code = module_obj[0]
                if biz_code in User_Alert_Dict.keys():
                    biz_code_value = User_Alert_Dict[biz_code]
                    self.alertDict['alert_dict']['module'] = {
                        'importance': biz_code_value['importance'],
                        'biz_code': biz_code
                    }
                    biz_code_list.append(biz_code)

                    for phone in biz_code_value['mobile'].split(','):
                        if phone not in mobile_list:
                            mobile_list.append(phone)

            if len(biz_code_list) > 0:
                self.alertDict['alert_dict']['biz_code_list'] = biz_code_list

            if len(mobile_list) > 0:
                self.alertDict['alert_dict']['mobile_list'] = mobile_list

            logging.info("钉钉推送消息的补充内容: {}".format(self.alertDict['alert_dict']))
            return True
        except:
            logging.error('数据正则校验失败')
            logging.error(traceback.format_exc())
            return True

    def run(self):
        check_key = ['message', 'fields']
        check_fields_key = ['server', 'host_ip']
        if self.data_queue is None or self.dding_url is None:
            logging.error('数据通道或钉钉连接未配置')
            return

        while True:
            try:
                if self.data_queue.empty():
                    time.sleep(0.5)
                    continue
                dataDict = self.data_queue.get()
                check_status = True

                for k in check_key:
                    if k not in dataDict:
                        logging.error('check_key 校验失败, 数据: {}'.format(dataDict))
                        check_status = False
                for k in check_fields_key:
                    if k not in dataDict['fields']:
                        logging.error('check_fields_key 校验失败, 数据: {}'.format(dataDict))
                        check_status = False

                if check_status:
                    self.alertDict['ip'] = dataDict['fields']['host_ip']
                    self.alertDict['project'] = dataDict['fields']['server']
                    logging.info('开始推送异常日志: {}-{}'.format(self.alertDict['project'], self.alertDict['ip']))
                    if len(dataDict['message']) > 200:
                        self.alertDict['msg'] = dataDict['message'][0:200]
                    else:
                        self.alertDict['msg'] = dataDict['message']

                    self.alertDict['complete_msg'] = dataDict['message']
                    for funcName in self.hook_func_list:
                        be_called_function = getattr(self, funcName, None)
                        if be_called_function is None:
                            logging.error('无{}钩子函数'.format(funcName))
                        else:
                            status = be_called_function()
                            if not status:
                                logging.error('{}钩子返回False'.format(funcName))
                                break
            except:
                logging.error('数据推送线程失败')
                logging.error(traceback.format_exc())
                self.send_alert()
                time.sleep(1)


class DataConsume(object):
    def __init__(self):
        self.kafka_url = []
        self.ding_ding_dict = {
            'project_test': {
                0: '',
                1: ''
            },
            'bao_dai': {
                0: '',
                1: ''
            },
            'gong_yin_lian': {
                0: '',
                1: ''
            }
        }

        self.hook_dict = {
            'project_test': ['frequency_control', 'save_error_log', 'alert_data_generation', 'ding_ding_push',
                             'is_holiday', 'frequency_control_sms', 'sms_push'],
            'bao_dai': ['frequency_control', 'save_error_log', 'alert_data_generation', 'ding_ding_push'],
            'gong_yin_lian': ['frequency_control', 'save_error_log', 'alert_data_generation', 'ding_ding_push'],
        }

        self.queue_dict = {}
        self.thread_dict = {}
        self.topic = ''
        self.group_id = ''
        self.kafka_consumer = None

    def init_service(self):
        try:
            for k, v in self.ding_ding_dict.items():
                queueObj = queue.Queue(100)
                threadObj = DDingPush(d_url=v, d_queue=queueObj, d_hook_list=self.hook_dict[k])
                threadObj.start()
                self.queue_dict[k] = queueObj
                self.thread_dict[k] = threadObj

            self.kafka_consumer = kafka.KafkaConsumer(
                self.topic,
                group_id=self.group_id,
                bootstrap_servers=self.kafka_url,
                auto_offset_reset='latest',
                max_poll_records=10,
                ssl_check_hostname=False,
                api_version=(0, 9),
                value_deserializer=lambda m: json.loads(m)
            )
            logging.info('初始化成功')
        except:
            logging.error('init_service 初始化失败')
            logging.error(traceback.format_exc())

    def consume_data(self):
        for message in self.kafka_consumer:
            if message is not None:
                # print("%s:%d:%d: key=%s value=%s" % (message.topic, message.partition,
                #                                      message.offset, message.key,
                #                                      message.value))
                logging.info('kafka消费位置: {}'.format(message.offset))
                try:
                    msg_dict = message.value
                    if 'fields' in msg_dict.keys():
                        if 'notice_type' in msg_dict['fields'].keys():
                            notice_type = msg_dict['fields']['notice_type']
                            if notice_type in self.queue_dict.keys():
                                self.queue_dict[notice_type].put(msg_dict)
                            else:
                                logging.error("无对应的通知通道: {}".format(notice_type))
                        else:
                            logging.error("数据缺少notice_type字段: {}".format(msg_dict))
                    else:
                        logging.error("数据缺少fields字段: {}".format(msg_dict))
                except:
                    logging.error('consume_data 消费kafka数据异常失败')
                    logging.error(traceback.format_exc())
            else:
                print("no message")

    def run(self):
        # 刷新用户告警字典
        generateAlertDict()
        json_generate_alert_dict()
        sms_alert_date_dict()
        logging.info("完成所有字典刷新")
        # 开始初始化数据
        self.init_service()
        logging.info("完成初始化数据")
        # 开始消费推送数据
        logging.info("开始消费推送数据")
        self.consume_data()


if __name__ == '__main__':
    http_thread = threading.Thread(target=app.run, kwargs={'host': '0.0.0.0', 'port': '7000'})
    http_thread.start()
    obj = DataConsume()
    obj.run()
